package com.cnlive.logpare.app

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import com.cnlive.logpare.util.IPSeekerExt
import scala.collection.mutable.HashMap
import java.net.URLDecoder
import com.cnlive.logpare.util.ParseLogUtil

import java.text.SimpleDateFormat
import java.util.Locale
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.LongType
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.SQLContext
import java.util.Properties
import java.sql.DriverManager
import java.sql.PreparedStatement
import java.sql.Connection
import org.apache.hadoop.hbase.client.HBaseAdmin
import org.apache.hadoop.hbase.HBaseConfiguration
import java.util.UUID
import java.util.Date
import com.cnlive.logpare.util.Constant

object ParseAppLog {
  val sdf=new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss",Locale.ENGLISH)
  val dateFormat=new SimpleDateFormat("yyyyMMddHHmmss",Locale.ENGLISH)
  val dateFormat3 =new SimpleDateFormat("yyyy-MM-dd", Locale.ENGLISH)
  def main(args: Array[String]): Unit = {
    var env=1
    if(args.length>0){
     env=args(0).toInt 
    }
    val conf = new SparkConf().setAppName("ParseAppLog")
    val nowTime=dateFormat.format(new Date).toLong
    var date=dateFormat3.format(new Date)
    if(env==1){
      conf.setMaster("local")
    }
    if(args.length>0){
      date=args(1)
    }
    val sc = new SparkContext(conf)
    val files=env match{
      case 1 => "file:///F:/工作资料/83app_access_201705081100.log"
      case _ => "hdfs://sxt/nginx_logs/app/offline/"+date+"/*"
    }
    val qqwryPath=env match{
        case 2 =>Constant.qqwryLocalServer
        case 3 =>Constant.qqwryServer
        case _=>""
    }
    val zk_addr=env match{
        case 3=>"jdbc:phoenix:node001,node002,node003:2181"
        case _=>"jdbc:phoenix:node180,node181,node182:2181"
    }
   val tabData= sc.textFile(files).mapPartitions(x=> {
      var result = List[String]()
      val sessionMap=new HashMap[String, Int]()
      sessionMap.put("count", 0);
      val props=new Properties();
		  val conn = DriverManager.getConnection(zk_addr, props);
		  val sql="""
		    upsert into APPLOG 
		    (NGINXTIME,HASHCODE,TM,UID,SPID,APPID,IP,VERSION,CITY,LOCATION,OPERATOR,PLAT,EVENTID,ISNEWUSER,ISSTART) 
		     values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
		   """
		  val prest = conn.prepareStatement(sql)  
      val iPSeekerExt=env match {
          case 1 =>  IPSeekerExt.getInit()
          case 2 =>  IPSeekerExt.getInit(true,qqwryPath)
          case 3 =>  IPSeekerExt.getInit(true,qqwryPath)
      }
		  var count=0
      while (x.hasNext) {
        try{
          
           val lineText=x.next()
           count=count+1
           result= result.::(parLineText(iPSeekerExt,lineText,sessionMap,prest,conn))
           sessionMap.put("count", count);
           }catch {
             case t: Throwable => t.printStackTrace()
           }
      }
      conn.commit()
      conn.close()
      result.iterator
    })
    .count()
  }
  val ts=""" \"(.*?)\"""".r
  def parLineText(iPSeekerExt:IPSeekerExt,lineText:String,sessionMap:HashMap[String, Int],prest:PreparedStatement,con:Connection):String={
     val mapRequest=new HashMap[String, String]()
     val ipandtime=ts.split(lineText).toList
     val urlandtype=ts.findAllIn(lineText).toList
     if(ipandtime.length==2&&urlandtype.length==3){//
        val ipandtimeArr=ipandtime(0).split(" ")//由于IP可能有多个所以
        val timeIndx=ipandtimeArr.length-2
        val ip=if (ipandtimeArr(0).indexOf(",") != -1) ipandtimeArr(0).substring(0, ipandtimeArr(0).indexOf(",")) else ipandtimeArr(0)
        val time=ipandtimeArr(timeIndx).replace("[", "")
        mapRequest.put("nginxtime", time)//时间
        val d=iPSeekerExt.analyticIp(ip)
        val city=d.getCity
        val province=d.getProvince
        val country=d.getCountry
        val operator=d.getOperator
        mapRequest.put("ip", ip);//城市
        mapRequest.put("location", province);//省份
        mapRequest.put("city", city);//城市
        mapRequest.put("operator", operator+"")//运营商
      val urlandper=urlandtype(0).split(" ")
      if(urlandper.length==4){//解析url
          URLRequest(urlandper(2),mapRequest,sessionMap)
      }else{
          println("url解析有误"+lineText)
       }
       platParse(urlandtype(2),mapRequest)//解析操作系统
     }else{//数据格式不对
       println("数据格式不对"+lineText)
       return ""
     }
     if(mapRequest.contains("appid")
         &&mapRequest.contains("nginxtime")
         &&mapRequest.contains("eventid")
         &&mapRequest.contains("ip")
         &&mapRequest.contains("tm")
         &&mapRequest.contains("uid")
         &&mapRequest.contains("location")//省份
         &&mapRequest.contains("plat")//操作系统
         &&mapRequest.contains("city")//城市
         &&mapRequest.contains("operator")//运营商
         &&mapRequest.contains("isnewuser")//是否新用户
         ){
       val appid=mapRequest.getOrElse("appid", "null")
       val spidandappid=appid.split("_")
       val nginxtime=dateFormat.format(sdf.parse(mapRequest.getOrElse("nginxtime", "20101010000000"))).toLong
       val city=mapRequest.getOrElse("city", "null")
       val location=mapRequest.getOrElse("location", "null")
       val plat=mapRequest.getOrElse("plat", "199").toInt
       val eventid=mapRequest.getOrElse("eventid", "0").toInt
       val version=mapRequest.getOrElse("version", "0")
       val uid=mapRequest.getOrElse("uid", "0")
       val ip=mapRequest.getOrElse("ip", "0")
       val tm=mapRequest.getOrElse("tm", "0").toLong
       val operator=mapRequest.getOrElse("operator", "0").toInt
       var isnewuser=mapRequest.getOrElse("isnewuser", "0").toInt
       val isstart=if(eventid==1300)1 else 0 
       if(spidandappid.length==2){
         //NGINXTIME  | UID  | SPID  | APPID  | IP  | VERSION  | CITY  | LOCATION  | OPERATOR  | PLAT  | EVENTID  | ISNEWUSER
           var index=1
          prest.setLong(index, nginxtime);index=index+1
          prest.setLong(index, lineText.hashCode());index=index+1
          prest.setLong(index, tm);index=index+1
          prest.setString(index, uid);index=index+1
          prest.setString(index, spidandappid(0));index=index+1
          prest.setString(index, appid);index=index+1
          prest.setString(index, ip);index=index+1
          prest.setString(index, version);index=index+1
          prest.setString(index, city);index=index+1
          prest.setString(index, location);index=index+1
          prest.setInt(index, operator);index=index+1
          prest.setInt(index, plat);index=index+1
          prest.setInt(index, eventid);index=index+1
          prest.setInt(index, isnewuser);index=index+1
          prest.setInt(index, isstart);index=index+1
          prest.execute()
          //if(sessionMap.getOrElse("count", 10)/100==0){
            con.commit()
          //}
          ""
       }else{
           //println("appid不是_分割的"+appid)
           ""
       }
     }else{
        println("appid为空")
        ""
     }
  }
  //解析操作系统
  def platParse(text:String,mapRequest:HashMap[String, String]){
    val d= ParseLogUtil.get_mobile_system(URLDecoder.decode(text,"utf-8").toLowerCase())
     mapRequest.put("plat",d+"")
  }
  //url解析
  def URLRequest(url:String,mapRequest:HashMap[String, String],sessionMap:HashMap[String, Int]){
   val strUrlParam=TruncateUrlPage(URLDecoder.decode(url,"utf-8").toLowerCase())
   if(strUrlParam.length!=0){
     val arrSplit=strUrlParam.split("&")
     for(strSplit<-arrSplit){
       val arrSplitEqual=strSplit.split("[=]")
       if(arrSplitEqual.length>1){
         mapRequest.put(arrSplitEqual(0), arrSplitEqual(1));
       }
     }
   }
   val uid=mapRequest.getOrElse("uid", "")
   val tm=mapRequest.getOrElse("tm", "")
   if((!uid.equals(""))&&(!tm.equals(""))){
     val uidAndtm=uid.split("_")
     if(uidAndtm.length==2){
     if(uidAndtm(0).length()==13&&uidAndtm(0).matches("^[0-9]*$")){//uid中包含1494210652000_j1tsu24521
         if(uidAndtm(0).toLong==tm.toLong){
           if(sessionMap.contains(uid)){
            mapRequest.put("isnewuser", "0");   
           }else{
              sessionMap.put(uid, 1);
              mapRequest.put("isnewuser", "1");   
           }
         }else{
            mapRequest.put("isnewuser", "0"); 
         }
       }else if(uidAndtm(1).length()==13&&uidAndtm(1).matches("^[0-9]*$")){
           if(uidAndtm(1).toLong==tm.toLong){
             if(sessionMap.contains(uid)){
               mapRequest.put("isnewuser", "0");   
             }else{
                sessionMap.put(uid, 1);
                mapRequest.put("isnewuser", "1");   
             }
           }else{
              mapRequest.put("isnewuser", "0"); 
           }
      }
     }else{
        mapRequest.put("isnewuser", "0"); 
     }
   }else{
     println("url参数解析不合格！")
   }
  }
  //截取URL
  def TruncateUrlPage(strURL:String):String={
    val arrSplit=strURL.split("[?]")
    var strAllParam="";
    if(strURL.length()>1){
      if(arrSplit.length>1){
          if(arrSplit(1)!=null){
           strAllParam=arrSplit(1)
          }
      }
    }
    strAllParam
  }
}